package cn.edu.thu.laud.thrift.customed;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.thrift.ProcessFunction;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TBase;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TMessage;
import org.apache.thrift.protocol.TMessageType;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TProtocolUtil;
import org.apache.thrift.protocol.TType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.hive.service.ThriftHive;
import cn.edu.thu.laud.hive.service.HiveServer.HiveServerHandler;
import cn.edu.thu.laud.server.thrift.LaUD;
import cn.edu.thu.laud.thrift.LaUDCassandraServer;
import cn.edu.thu.laud.thrift.LaUDServer;
import cn.edu.thu.laud.thrift.service.LaudIfaceImpl;

// 
@SuppressWarnings("rawtypes")
public class CustomedLaUDProcessor extends org.apache.thrift.TBaseProcessor implements org.apache.thrift.TProcessor{
	private static Map<String, ProcessFunction<Cassandra.Iface,? extends TBase> >cassandraProcessFunctionMap=new HashMap<String, ProcessFunction<Cassandra.Iface,? extends TBase>>();
	private static Map<String, ProcessFunction<LaUD.Iface,? extends TBase>  >  laUDProcessFunctionMap=new   HashMap<String, ProcessFunction<LaUD.Iface,? extends TBase>>();
	//hive is special .
	private Map<String, ProcessFunction<ThriftHive.Iface,? extends TBase> > hiveProcessFunctionMap=new HashMap<String, ProcessFunction<ThriftHive.Iface,? extends TBase> >();
	private   final Cassandra.Iface cassandraIface;
	private   final LaUD.Iface laudIface;
	private final HiveConf hiveConf;

	static Logger logger = LoggerFactory.getLogger(CustomedLaUDProcessor.class);
	
//	final LaudIfaceImpl laudImpl= new LaudIfaceImpl(conf,((LaUDCassandraServer)cassandraServer).getThreadLocalClientStateField());
//	
//	laudImpl.initCassandraIface(cassandraServer);
	private  ThriftHive.Iface hiveIface;
	static {
		try {
			//cassandra thrift
			Method method= new Cassandra.Processor<Cassandra.Iface>(null).getClass().getDeclaredMethod("getProcessMap", new Class[]{Map.class});
			method.setAccessible(true);
			method.invoke(null,cassandraProcessFunctionMap);
			//laud thrift
			method= new LaUD.Processor<LaUD.Iface>(null).getClass().getDeclaredMethod("getProcessMap", new Class[]{Map.class});
			method.setAccessible(true);
			method.invoke(null,laUDProcessFunctionMap);
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (NoSuchMethodException e) {
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		} catch (InvocationTargetException e) {
			e.printStackTrace();
		}		 
	}	
	@SuppressWarnings("unchecked")
	public CustomedLaUDProcessor(Cassandra.Iface ci,HiveConf hiveconf) {
		super(null,null);
		this.cassandraIface=ci;
		LaudIfaceImpl laudIfaceImpl=new LaudIfaceImpl(hiveconf,((LaUDCassandraServer)cassandraIface).getThreadLocalClientStateField());
		laudIfaceImpl.initCassandraIface(cassandraIface);
		this.laudIface=laudIfaceImpl;
		this.hiveConf=hiveconf;
		this.hiveIface=null;
	}

/**
 * BUG: 当客户端退出或者非正常退出时，没有删除其他地方引用的ClientState.(LaUDDataServer)
 */
	@SuppressWarnings("unchecked")
	@Override
	public boolean process(TProtocol in, TProtocol out) throws TException {
		TMessage msg = in.readMessageBegin();
		ProcessFunction fn=null;
		//because we use jdbc first,so we put laudProcess first.
		//if we want to use thrift first ,we'd better put cassandraProcess first..
		if((fn=laUDProcessFunctionMap.get(msg.name))!=null){
			if(((LaudIfaceImpl)laudIface).getProcessor()==null){
				((LaudIfaceImpl)laudIface).setProcessor(this);
			}
			fn.process(msg.seqid, in, out, laudIface);
		}else if((fn=cassandraProcessFunctionMap.get(msg.name))!=null){
			fn.process(msg.seqid, in, out, cassandraIface);
		}else {
			if(LaUDServer.includeHive){
				if(hiveIface==null){
					this.setHiveIface(null);
				}
				if((fn=hiveProcessFunctionMap.get(msg.name))!=null){
					try {
						if(hiveIface==null){
							hiveIface = new HiveServerHandler(hiveConf);
							logger.info("create a new hiveServerHandler");
						}
						fn.process(msg.seqid, in, out, hiveIface);
					} catch (MetaException e) {
						throw new TException(e);
					}
				} else{//(fn == null)
					TProtocolUtil.skip(in, TType.STRUCT);
					in.readMessageEnd();
					TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");
					out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));
					x.write(out);
					out.writeMessageEnd();
					out.getTransport().flush();
					return true;
				} 
			}else{
				TProtocolUtil.skip(in, TType.STRUCT);
				in.readMessageEnd();
				TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");
				out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));
				x.write(out);
				out.writeMessageEnd();
				out.getTransport().flush();
				return true;
			}
		}
		return true;
	}


	public ThriftHive.Iface getHiveIface() {
		return hiveIface;
	}
	
	public void setHiveIface(ThriftHive.Iface hiveIface) {
		this.hiveIface=hiveIface;
		if(hiveProcessFunctionMap.size()==0){
			Method method;
			try {
				method = new ThriftHive.Processor<ThriftHive.Iface>(getHiveIface()).getClass().getDeclaredMethod("getProcessMap", new Class[]{Map.class});
				method.setAccessible(true);
				method.invoke(null,hiveProcessFunctionMap);
			} catch (SecurityException e) {
				e.printStackTrace();
			} catch (NoSuchMethodException e) {
				e.printStackTrace();
			} catch (IllegalArgumentException e) {
				e.printStackTrace();
			} catch (IllegalAccessException e) {
				e.printStackTrace();
			} catch (InvocationTargetException e) {
				e.printStackTrace();
			}	
		}
	}
	public Cassandra.Iface getCassandraIface(){
		return this.cassandraIface;
	}
}
